package edu.ustb.efws.pipeline.collect;

import edu.ustb.efws.EfwsProperties;
import edu.ustb.efws.base.domain.Detector;
import edu.ustb.efws.base.service.DetectorService;
import edu.ustb.efws.base.service.PositionService;
import edu.ustb.efws.base.service.SmsService;
import edu.ustb.efws.base.service.UserService;
import edu.ustb.efws.meter.service.MeterService;
import edu.ustb.efws.monitor.service.DetectorMonitorService;
import edu.ustb.efws.pipeline.collect.handler.CollectChina;
import edu.ustb.efws.pipeline.collect.cache.MaxICacheService;
import edu.ustb.efws.pipeline.collect.handler.meter.MeterPointPool;
import edu.ustb.efws.pipeline.service.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.time.Duration;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * 采集引擎
 *
 * @author TS Group
 */
@Component
public class CollectLauncher implements DisposableBean {
    private static final Logger LOGGER = LoggerFactory.getLogger(CollectLauncher.class);
    private static final AtomicBoolean RUNNING = new AtomicBoolean(false);

    private final EfwsProperties properties;
    private final CollectWorker[] workers;
    private final ScheduledExecutorService executorService;
    private final MeterPointPool pool;

    private final List<Detector> detectors;
    private final MaxICacheService maxICacheService;
    private final Object monitor = new Object();

    private volatile ScheduledFuture<?> launcherFuture;

    /**
     * 构造{@link CollectLauncher}
     *
     * @param properties {@link EfwsProperties}
     * @param collectService {@link CollectService}
     * @param detectorService {@link DetectorService}
     * @param iptLogService {@link IptLogService}
     * @param thosLogService {@link ThosLogService}
     * @param kyjLogService {@link KyjLogService}
     * @param positionService {@link PositionService}
     * @param riskLogService {@link RiskLogService}
     * @param meterService {@link MeterPointPool}
     * @param iptMaxIService {@link IptMaxIService}
     * @param userService {@link UserService}
     * @param smsService {@link SmsService}
     */
    @Autowired
    public CollectLauncher(EfwsProperties properties, CollectService collectService,
                           DetectorService detectorService, IptLogService iptLogService,
                           ThosLogService thosLogService, KyjLogService kyjLogService,
                           PositionService positionService, RiskLogService riskLogService,
                           MeterService meterService,
                           UserService userService, SmsService smsService,
                           DetectorMonitorService detectorMonitorService,
                           MaxICacheService maxICacheService){

        this.properties = properties;
        this.executorService = Executors.newScheduledThreadPool(2);
        this.pool = new MeterPointPool(properties, meterService);
        this.detectors = detectorService.all();
        detectorMonitorService.init(detectors);
        CollectChina collectChina = new CollectChina(iptLogService, thosLogService,
                kyjLogService, positionService, riskLogService, pool, properties,
                userService, smsService, detectorMonitorService, maxICacheService);

        this.workers = buildWorkers(properties.getCollectWorks(), collectChina, collectService);

        this.maxICacheService = maxICacheService;
    }

    /**
     * 构建采集工作者
     *
     * @param count 工作者数量
     * @param collectChina {@link CollectChina}
     * @param collectService {@link CollectService}
     * @return 采集工作者数组
     */
    private CollectWorker[] buildWorkers(int count, CollectChina collectChina, CollectService collectService){

        ThreadFactory threadFactory = new CollectThreadFactory();
        CollectWorker[] workers = new CollectWorker[count];
        for(int i = 0; i < count; i++){
            workers[i] = new CollectWorker(collectChina, collectService, threadFactory);
        }

        return workers;
    }

    //开始数据采集
    public void start(){
        if(properties.isEnableCollect() && RUNNING.compareAndSet(false, true)){
            //生成采集信号
            launcherFuture = executorService.scheduleAtFixedRate(
                    this::launch,
                    Duration.ofSeconds(5).toMillis(),
                    properties.getCollectPeriodMills(),
                    TimeUnit.MILLISECONDS);

            //保存测量指标
            executorService.scheduleAtFixedRate(
                    pool::saveUpdate,
                    Duration.ofSeconds(30).toMillis(),
                    Duration.ofSeconds(30).toMillis(),
                    TimeUnit.MILLISECONDS);

            LOGGER.info("Collect launcher start ...");
        }
    }

    private void launch(){
        int count = workers.length;
        try{
            //分发采集任务
            for(Detector detector: detectors){
                workers[detector.getId() % count].push(detector);
            }
        }catch (InterruptedException e){
            LOGGER.error("Collect launcher interrupted error={}", e.getMessage());

            if(launcherFuture != null){
                launcherFuture.cancel(true);
                LOGGER.info("Stop collect launcher ...");
            }

            Thread.currentThread().interrupt();
        }
    }

    @Override
    public void destroy() {
        if(RUNNING.get()){
            //停止分发服务
            executorService.shutdown();
            //休眠等待任务处理完成
            waitWorkerShutdown();

            pool.saveAll();

            maxICacheService.scheduledSyncDB();;

            LOGGER.info("Shutdown collect launcher ...");
        }
    }

    private void waitWorkerShutdown(){
        try{
            synchronized (monitor){
                monitor.wait(Duration.ofSeconds(properties.getWaitShutdownSec()).toMillis());
            }

            for(CollectWorker worker: workers){
                worker.shutdown();
            }
        }catch (InterruptedException e){
            //None instance
        }
    }
}
